package com.sunzm.spark.streaming.callcenter

import java.sql.DriverManager

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.{DateFormatUtils, DateUtils}
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Duration, Seconds, State, StateSpec, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream, MapWithStateDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.{KafkaUtils, LocationStrategies}

/**
 * 呼叫中心客服工作量统计-V2
 *
 * 使用mapWithState保存同一个通话的状态数据（通话开始时间、振铃时间、接听时间，挂机时间、结束时间等）
 * 从而时长通话时长类指标的统计
 *
 * 问题：
 * 1，修改代码后，不删除checkpoint，程序无法启动或者执行的是旧代码
 * 2，程序自动维护kafka的偏移量，无法保证精确一次语义（同一条消息仅且消费一次，因为这种方式是周期性提交偏移量，
 * 假设10秒提交一次偏移量，在第8秒的时候，程序挂掉了。处理了8秒内接收到的数据，但是偏移量还没来得及提交，下次重启，这8秒的数据
 * 会被重复消费）
 */
object CallCenterStaffJobV2 {
  //状态数据需要保存在 checkpointDirectory 目录中
  private val checkpointDirectory = "data/spark/streaming/ck/20210620001/"

  def main(args: Array[String]): Unit = {
    val ssc = StreamingContext.getOrCreate(checkpointDirectory, creatingFunc)

    //启动程序
    ssc.start()
    ssc.awaitTermination()
  }

  def creatingFunc() = {

    val sparkConf = new SparkConf()
      .setAppName(s"${this.getClass.getSimpleName.stripSuffix("$")}")
      .setMaster("local[*]")

    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //设置checkpoint目录
    ssc.checkpoint(checkpointDirectory)

    val kafkaParams = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "82.156.210.70:9093",
      ConsumerConfig.GROUP_ID_CONFIG -> "group_id_call_center",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer])

    val topics = Array("call-center-message")
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //把消息中的value取出来（不需要key）
    val kafkaMessageDStream: DStream[String] = kafkaDStream.map(_.value)

    //开始进行处理
    processDStream(kafkaMessageDStream)

    ssc
  }

  def processDStream(kafkaMessageDStream: DStream[String]) = {

    //变成key-value的类型，key是通话Id，value是一行数据, 因为只有K-V类型才能调用mapWithState算子
    val pairsDStream: DStream[(String, String)] = kafkaMessageDStream.transform(rdd => {

      rdd.filter(line => {
        //先把格式异常的数据过滤掉
        var res = false

        try {
          val jSONObject = JSON.parseObject(line)

          if (jSONObject.containsKey("companyId") &&
            jSONObject.containsKey("staffId") &&
            jSONObject.containsKey("dateTime")) {
            res = true
          } else {
            println(s"缺少字段[companyId,staffId,dateTime]: ${line}")
          }
        } catch {
          case _ => {
            println(s"JSON格式异常: ${line}")
          }
        }

        res
      }).map(line => {
        //解析JSON
        val jSONObject = JSON.parseObject(line)

        //通话ID
        val callId = jSONObject.getString("callId")

        (callId, line)
      })

    })

    val spec: StateSpec[String, String, String, Option[String]] = StateSpec.function(mappingFunc)

    val mapWithStateDStream: MapWithStateDStream[String, String, String, Option[String]] = pairsDStream.mapWithState(spec)

    //开始计算
    val callResultDS = mapWithStateDStream.transform(rdd => {
      rdd.filter(_.nonEmpty).map(lineOption => {
        val line = lineOption.get

        //解析JSON
        val jSONObject = JSON.parseObject(line)

        //公司ID
        val companyId = jSONObject.getString("companyId")
        //坐席ID
        val staffId = jSONObject.getString("staffId")
        //坐席名称
        val staffName = jSONObject.getString("staffName")
        //获取通话开始时间
        val startTime = jSONObject.getLongValue("startTime")
        //通话结束时间
        val endTime = jSONObject.getLongValue("endTime")
        //转换成 yyyy-MM-dd 格式
        val resultDate: String = DateFormatUtils.format(startTime, "yyyy-MM-dd")

        //开始统计
        //呼入方向： 0呼入 1呼出
        val direction: Int = jSONObject.getIntValue("direction")

        //呼入电话数量
        var callInCount = 0
        //呼入成功数量
        var callInSuccCount = 0
        //呼入通话时长
        var callInDuration = 0L
        //呼入振铃时长
        var callInRingDuration = 0L

        //呼出电话数量
        var callOutCount = 0
        //呼出成功数量
        var callOutSuccCount = 0
        //呼出通话时长
        var callOutDuration = 0L
        //呼出振铃时长
        var callOutRingDuration = 0L

        if (direction == 0) {
          //呼入
          //呼入电话数量
          callInCount = 1
          //呼入通话时长
          callInDuration = endTime - startTime

          //坐席接听时间
          val staffAnswerTime = jSONObject.getLongValue("staffAnswerTime")
          if (staffAnswerTime > 0) {
            //呼入，坐席接听，就表示呼入成功
            callInSuccCount = 1
            //坐席振铃时间
            val staffRingTime = jSONObject.getLongValue("staffRingTime")
            //呼入振铃时长
            callInRingDuration = staffAnswerTime - staffRingTime
          }
        } else {
          // 呼出
          //呼出电话数量
          callOutCount = 1
          //呼出通话时长
          callOutDuration = endTime - startTime

          //用户接听时间
          val userAnswerTime = jSONObject.getLongValue("userAnswerTime")
          if (userAnswerTime > 0) {
            //呼出，用户接听，就表示呼出成功
            callOutSuccCount = 1
            //用户振铃时间
            val userRingTime = jSONObject.getLongValue("userRingTime")
            //呼出振铃时长
            callOutRingDuration = userAnswerTime - userRingTime
          }
        }

        ((resultDate, companyId, staffId),
          (callInCount, callInSuccCount, callInRingDuration, callInDuration,
            callOutCount, callOutSuccCount, callOutRingDuration, callOutDuration, staffName))

      }).reduceByKey((t1, t2) => {

        val callInCount: Int = t1._1 + t2._1
        val callInSuccCount: Int = t1._2 + t2._2
        val callInRingDuration: Long = t1._3 + t2._3
        val callInDuration: Long = t1._4 + t2._4
        val callOutCount: Int = t1._5 + t2._5
        val callOutSuccCount: Int = t1._6 + t2._6
        val callOutRingDuration: Long = t1._7 + t2._7
        val callOutDuration: Long = t1._8 + t2._8
        val staffName1: String = t1._9
        val staffName2: String = t2._9

        val staffName: String = if (StringUtils.isNotBlank(staffName1)) staffName1 else staffName2

        (callInCount, callInSuccCount, callInRingDuration, callInDuration, callOutCount,
          callOutSuccCount, callOutRingDuration, callOutDuration, staffName)
      })
    })

    //callResultDS.print()
    //将统计结果保存到MySQL中
    callResultDS.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        rdd.foreachPartition(p => {
          //获取数据库连接
          //获取数据库连接, 如果在foreach中获取连接，有多少条数据就要获取几次数据库连接
          //而在foreachPartition中，有几个分区就获取几次连接
          //数据库连接是很宝贵的，应该尽量减少获取数据库连接的次数
          val conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/mydb?useSSL=false", "root", "root")

          val sqlStr =
            """
              |INSERT INTO `cc_staff_job_status` (
              |	`resultDate`,
              |	`companyId`,
              |	`staffId`,
              |	`staffName`,
              |	`call_in_count`,
              |	`call_in_succ_count`,
              |	`call_in_ring_duration`,
              |	`call_in_duration`,
              |	`call_out_count`,
              |	`call_out_succ_count`,
              |	`call_out_ring_duration`,
              |	`call_out_duration`
              |	) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
              | ON DUPLICATE KEY UPDATE
              |   `staffName` = ?,
              |	`call_in_count` = `call_in_count` + ?,
              |	`call_in_succ_count` = `call_in_succ_count` + ?,
              |	`call_in_ring_duration` = `call_in_ring_duration` + ?,
              |	`call_in_duration` = `call_in_duration` + ?,
              |	`call_out_count` = `call_out_count` + ?,
              |	`call_out_succ_count` = `call_out_succ_count` + ?,
              |	`call_out_ring_duration` = `call_out_ring_duration` + ?,
              |	`call_out_duration` = `call_out_duration` + ?
              |""".stripMargin

          val pstmt = conn.prepareStatement(sqlStr)

          //开始真正插入数据
          var currentBatchSize = 0
          var startTime = System.currentTimeMillis()
          p.foreach {
            case ((resultDate, companyId, staffId),
            (callInCount, callInSuccCount, callInRingDuration, callInDuration,
            callOutCount, callOutSuccCount, callOutRingDuration, callOutDuration, staffName)) => {

              pstmt.setString(1, resultDate)
              pstmt.setString(2, companyId)
              pstmt.setString(3, staffId)
              pstmt.setString(4, staffName)
              pstmt.setString(5, callInCount + "")
              pstmt.setString(6, callInSuccCount + "")
              pstmt.setString(7, callInRingDuration + "")
              pstmt.setString(8, callInDuration + "")
              pstmt.setString(9, callOutCount + "")
              pstmt.setString(10, callOutSuccCount + "")
              pstmt.setString(11, callOutRingDuration + "")
              pstmt.setString(12, callOutDuration + "")
              pstmt.setString(13, staffName)
              pstmt.setString(14, callInCount + "")
              pstmt.setString(15, callInSuccCount + "")
              pstmt.setString(16, callInRingDuration + "")
              pstmt.setString(17, callInDuration + "")
              pstmt.setString(18, callOutCount + "")
              pstmt.setString(19, callOutSuccCount + "")
              pstmt.setString(20, callOutRingDuration + "")
              pstmt.setString(21, callOutDuration + "")

              pstmt.execute()

              //添加到一个批次
             /* pstmt.addBatch()
              currentBatchSize += 1

              //执行的时长(毫秒)
              val useTime = System.currentTimeMillis() - startTime
              if(currentBatchSize >= 20 || useTime >= 5 * 1000){
                //SQL语句数量达到20条或者时间相隔5秒，触发一次

                //执行这个批次的所有sql
                pstmt.executeBatch()
                //清空批次
                pstmt.clearBatch()
              }*/

            }
          }

          //关闭连接
          if (pstmt != null) {
            pstmt.close()
          }

          if (conn != null) {
            conn.close()
          }

        })
      }
    })
  }


  /**
   * 记录和处理状态数据的方法
   */
  val mappingFunc = (callId: String, line: Option[String], state: State[String]) => {

    if (line.nonEmpty) {
      //获取状态数据
      val callStateStr = state.getOption().getOrElse("{}")

      val callStateJSON: JSONObject = JSON.parseObject(callStateStr)

      //解析JSON
      val jsonStr = line.get
      val jSONObject = JSON.parseObject(jsonStr)

      //公司ID
      val companyId = jSONObject.getString("companyId")
      //坐席ID
      val staffId = jSONObject.getString("staffId")
      //时间 yyyyMMddHHmmssSSS -> yyyy-MM-dd
      val dateTimeStr = jSONObject.getString("dateTime")
      //对字符串的日期进行格式转换
      val dateTimeLong: Long = DateUtils.parseDate(dateTimeStr, "yyyyMMddHHmmssSSS").getTime

      //呼入方向： 0呼入 1呼出
      val direction: Int = jSONObject.getIntValue("direction")

      /**
       * 消息类型:
       * start通话开始; staff-ring坐席振铃; staff-answer坐席接听;
       * staff-hunup坐席挂机; user-ring：用户振铃； user-answer：用户接听;
       * user-hunup：用户挂断; stop:通话结束
       */
      val messageType: String = StringUtils.trim(jSONObject.getString("messageType"))

      messageType match {
        case "start" => {
          //通话开始
          //记录通话开始时间
          callStateJSON.put("startTime", dateTimeLong)
        }
        case "staff-ring" => {
          //坐席振铃
          callStateJSON.put("staffRingTime", dateTimeLong)
        }
        case "staff-answer" => {
          //坐席接听
          callStateJSON.put("staffAnswerTime", dateTimeLong)
        }
        case "staff-hunup" => {
          //坐席挂掉
          callStateJSON.put("staffHunupTime", dateTimeLong)
        }
        case "user-ring" => {
          //用户振铃
          callStateJSON.put("userRingTime", dateTimeLong)
        }
        case "user-answer" => {
          //用户接听
          callStateJSON.put("userAnswerTime", dateTimeLong)
        }
        case "user-hunup" => {
          //用户挂段
          callStateJSON.put("userHunupTime", dateTimeLong)
        }
        case "stop" => {
          //通话结束
          callStateJSON.put("endTime", dateTimeLong)
        }
        case _ => {
          println(s"未知的消息类型: ${messageType}")
        }
      }

      //记录呼叫方向
      callStateJSON.put("direction", direction)
      //记录坐席ID和公司ID
      if (StringUtils.isNotBlank(companyId)) {
        callStateJSON.put("companyId", companyId)
      }
      if (StringUtils.isNotBlank(staffId)) {
        callStateJSON.put("staffId", staffId)
      }

      //获取通话结束时间
      val endTime = callStateJSON.getLongValue("endTime")
      if (endTime > 0) {
        //通话结束，清空状态数据
        state.remove()
        //输出结果
        Some(callStateJSON.toJSONString)
      } else {
        //更新状态数据
        state.update(callStateJSON.toJSONString)
        //不输出
        None
      }
    } else {
      //不输出
      None
    }

  }
}
